package com.yycx.msg.provider.websocket;

import com.alibaba.fastjson.util.TypeUtils;
import com.baomidou.mybatisplus.core.toolkit.IdWorker;
import com.yycx.common.base.entity.EntityMap;
import com.yycx.common.base.service.WebSocketMsgService;
import com.yycx.common.base.utils.FlymeUtils;
import com.yycx.common.constants.CommonConstants;
import com.yycx.common.constants.QueueConstants;
import com.yycx.common.utils.DateUtils;
import com.yycx.common.utils.JsonUtils;
import com.yycx.common.utils.SpringContextHolder;
import com.yycx.module.admin.provider.service.BaseUserService;
import com.yycx.msg.client.entity.MsgContent;
import com.yycx.msg.client.entity.MsgReceive;
import com.yycx.msg.client.entity.MsgType;
import com.yycx.msg.provider.service.MsgContentService;
import com.yycx.msg.provider.service.MsgReceiveService;
import com.yycx.msg.provider.service.MsgTypeService;
import com.yycx.starter.rabbitmq.client.RabbitMqClient;
import com.yycx.starter.rabbitmq.config.RabbitComponent;
import com.rabbitmq.client.Channel;
import lombok.extern.slf4j.Slf4j;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.amqp.support.AmqpHeaders;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.handler.annotation.Header;
import org.springframework.security.crypto.password.PasswordEncoder;
import org.springframework.stereotype.Component;

import javax.annotation.PostConstruct;
import javax.websocket.*;
import javax.websocket.server.PathParam;
import javax.websocket.server.ServerEndpoint;
import java.io.IOException;
import java.io.Serializable;
import java.util.*;
import java.util.concurrent.CopyOnWriteArraySet;

/**
 * @author
 */
@ServerEndpoint("/websocket/{userId}/{sendType}")//此注解相当于设置访问URL
@Component
@Slf4j
public class WebSocketServerEndpoint implements WebSocketMsgService {

    @Autowired
    public RabbitMqClient rabbitMqClient;
    @Autowired
    public MsgContentService msgContentService;
    @Autowired
    public MsgReceiveService msgReceiveService;
    @Autowired
    public MsgTypeService msgTypeService;

    private BaseUserService baseUserService;

    /**
     * 心跳消息
     */
    private static final String heartMessage = "ping";

    /**
     * concurrent包的线程安全Set，用来存放每个客户端对应的WebSocketEntity对象。
     */
    private static final Set<WebSocketEntity> connections = new CopyOnWriteArraySet<>();


    /**
     * 建立连接成功
     *
     * @param session
     * @param userId
     */
    @OnOpen
    public void onOpen(Session session, @PathParam(value = "userId") String userId, @PathParam(value = "sendType") String sendType) {
        try {
            addUser(userId, sendType, session);
            if (FlymeUtils.isEmpty(baseUserService)) {
                baseUserService = SpringContextHolder.getBean("baseUserServiceImpl");
            }
            baseUserService.setOnLine(userId, CommonConstants.ENABLED);
        } catch (Exception e) {
        }
    }

    /**
     * 连接关闭
     */
    @OnClose
    public void onClose(@PathParam("userId") String userId, @PathParam("sendType") String sendType, CloseReason reason) {
        removeUser(userId, sendType, reason);
        if (FlymeUtils.isEmpty(baseUserService)) {
            baseUserService = SpringContextHolder.getBean("baseUserServiceImpl");
        }
        if (FlymeUtils.isNotEmpty(baseUserService)) {
            //baseUserService.setOnLine(userId, CommonConstants.DISABLED);
        }

    }

    /**
     * 接收客户端消息
     *
     * @param message
     */
    @OnMessage
    public void onMessage(String message) {
        if (FlymeUtils.isNotEmpty(message) && !message.equals(heartMessage)) {
            pushMessage(message);
        }
    }

    /**
     * 服务端推送消息
     *
     * @param userId
     * @param message
     */
    public void pushMessage(String userId, String sendType, String message) {
        WebSocketEntity entity = getUserEntity(userId, sendType);
        Session session = entity.getSession();
        if (session != null && session.isOpen()) {
            try {
                log.info("【websocket消息】 单点消息:" + message);
                session.getAsyncRemote().sendText(message);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 服务端推送消息
     *
     * @param userId
     * @param sendType
     * @param map
     */
    public void pushMessage(String userId, String sendType, Map<String, Object> map) {
        List<String> userIds = new ArrayList<>();
        userIds.add(userId);
        boolean save = initMsgData(userIds, sendType, map);
        WebSocketEntity entity = getUserEntity(userId, sendType);
        Session session = entity.getSession();
        initTypeCode(map);
        String jsonStr = JsonUtils.beanToJson(map);
        if (session != null && session.isOpen()) {
            try {
                log.info("【websocket消息】 单点消息:" + jsonStr);
                if (save) {
                    session.getAsyncRemote().sendText(jsonStr);
                }
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 服务端推送消息
     *
     * @param userId
     * @param sendType
     * @param map
     */
    public void pushMessage(EntityMap entityMap, String userId, String sendType) {
        Map<String, Object> map = new HashMap<>();
        map.putAll(entityMap);
        pushMessage(map, userId, sendType);
    }

    /**
     * 服务端推送消息
     *
     * @param userId
     * @param sendType
     * @param map
     */
    public void pushMessage(Map<String, Object> map, String userId, String sendType) {
        WebSocketEntity entity = getUserEntity(userId, sendType);
        Session session = entity.getSession();
        map.put("typeCode", sendType);
        String jsonStr = JsonUtils.beanToJson(map);
        if (session != null && session.isOpen()) {
            try {
                log.info("【websocket消息】 单点消息:" + jsonStr);
                session.getAsyncRemote().sendText(jsonStr);
            } catch (Exception e) {
                e.printStackTrace();
            }
        }
    }

    /**
     * 服务端推送消息
     *
     * @param userIdList
     * @param map
     */
    public void pushMessage(List<String> userIdList, String sendType, Map<String, Object> map) {
        boolean save = initMsgData(userIdList, sendType, map);
        for (String userId : userIdList) {
            WebSocketEntity entity = getUserEntity(userId, sendType);
            Session session = entity.getSession();
            initTypeCode(map);
            String jsonStr = JsonUtils.beanToJson(map);
            if (session != null && session.isOpen()) {
                try {
                    log.info("【websocket消息】 单点消息:" + jsonStr);
                    if (save) {
                        session.getAsyncRemote().sendText(jsonStr);
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
            }
        }
    }

    /**
     * 服务器端推送消息
     */
    public void pushMessage(String message) {
        try {
            connections.forEach(ws -> ws.getSession().getAsyncRemote().sendText(message));
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 服务器端推送消息(全推)
     */
    public void pushMessage(Map<String, Object> map, String sendType) {
        try {
            String jsonStr = JsonUtils.beanToJson(map);
            List<String> userIds = new ArrayList<>();
            connections.forEach(ws -> {
                String userId = ws.getUserId();
                userIds.add(userId);
                ws.getSession().getAsyncRemote().sendText(jsonStr);
            });
            initMsgData(userIds, sendType, map);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    /**
     * 后台发送消息到rabbitMQ
     *
     * @param msgContent
     * @param receiveUserIdList
     * @param sendType          系统公告1，个人消息2
     */
    public void sendMessage(MsgContent msgContent, List<Long> receiveUserIdList, Integer sendType) {
        log.info("【websocket消息】广播消息:" + msgContent.getMsgContent());
        Map<String, Object> map = JsonUtils.beanToMap(msgContent);
        map.put("receiveUserIds", receiveUserIdList);
        map.put("sendType", sendType);
        rabbitMqClient.sendMessage(QueueConstants.QUEUE_WEBSOCKET_MSG, map, 2);
    }

    /**
     * 后台发送消息到rabbitMQ
     *
     * @param msgTypeId
     * @param receiveUserIdList
     * @param sendType          系统公告1，个人消息2
     */
    public void sendMessage(Long msgTypeId, String title, String content, List<Long> receiveUserIdList, Integer sendType) {
        MsgContent msgContent = new MsgContent();
        msgContent.setMsgTypeId(msgTypeId);
        msgContent.setMsgContent(content);
        msgContent.setMsgId(IdWorker.getId());
        msgContent.setDeleted(CommonConstants.DEL_NO);
        msgContent.setMsgTitle(title);
        log.info("【websocket消息】广播消息:" + msgContent.getMsgContent());
        Map<String, Object> map = JsonUtils.beanToMap(msgContent);
        map.put("receiveUserIds", receiveUserIdList);
        map.put("sendType", sendType);
        EntityMap result = new EntityMap();
        result.putAll(map);
        rabbitMqClient.sendMessage(QueueConstants.QUEUE_WEBSOCKET_MSG, result);
    }


    /**
     * 后台发送消息到rabbitMQ
     *
     * @param msgTypeId
     * @param userId
     * @param sendType  系统公告1，个人消息2
     */
    public void sendMessage(Map<String, Object> data, String userId, String sendType) {
        pushMessage(data, userId, sendType);
    }

    /**
     * 此为单点消息
     *
     * @param msgContent
     * @param userId
     * @param sendType   系统公告1，个人消息2
     */
    public void sendMessage(MsgContent msgContent, String userId, Integer sendType) {
        Map<String, Object> map = JsonUtils.beanToMap(msgContent);
        map.put("receiveUserIds", userId);
        map.put("sendType", sendType);
        rabbitMqClient.sendMessage(QueueConstants.QUEUE_WEBSOCKET_MSG, map, 2);
    }


    /**
     * 记录信息到数据库
     *
     * @param userId
     * @param sendType
     * @param map
     */
    public Boolean initMsgData(List<String> userIds, String sendType, Map<String, Object> map) {
        String json = JsonUtils.beanToJson(map);
        MsgContent msgContent = JsonUtils.jsonToBean(json, MsgContent.class);
        List<MsgReceive> list = new ArrayList<>();
        for (String id : userIds) {
            Long msgId = msgContent.getMsgId();
            MsgReceive msgReceive = new MsgReceive();
            msgReceive.setMsgId(msgId);
            msgReceive.setReceiveUserId(Long.parseLong(id));
            msgReceive.setDeleted(CommonConstants.DEL_NO);
            msgReceive.setReadState(CommonConstants.DISABLED);
            msgReceive.setCreateTime(DateUtils.getNowDateTime());
            list.add(msgReceive);
        }
        boolean save = msgReceiveService.saveBatch(list);
        if (save) {
            if (sendType.equals("2")) {
                msgContent.setCreateTime(DateUtils.getNowDateTime());
                save = msgContentService.save(msgContent);
            }
        }
        return save;
    }

    /**
     * initTypeCode
     *
     * @param map
     */
    public Map<String, Object> initTypeCode(Map<String, Object> map) {
        Long msgTypeId = TypeUtils.castToLong(map.get("msgTypeId"));
        MsgType msgType = msgTypeService.getById(msgTypeId);
        map.put("typeCode", "");
        if (FlymeUtils.isNotEmpty(msgType)) {
            map.put("typeCode", msgType.getTypeCode());
        }
        return map;
    }

    /**
     * 用户上线
     *
     * @param userId
     * @param sendType
     * @param session
     */
    private synchronized void addUser(String userId, String sendType, Session session) {
        //查询当前用户是否已在集合
        WebSocketEntity entity = getUserEntity(userId, sendType);
        if (null == entity) {
            sendType = FlymeUtils.isEmpty(sendType) ? "" : sendType;
            WebSocketEntity webSocket = new WebSocketEntity(userId, sendType, session);
            connections.add(webSocket);
        } else {
            log.info("用户" + userId + "上线了，当前人数为:" + getUserOnlineNum());
        }

    }

    /**
     * 用户下线
     */
    private void removeUser(String userId, String sendType, CloseReason reason) {
        WebSocketEntity entity = getUserEntity(userId, sendType);
        if (null != entity) {
            try {
                if (entity.getSession().isOpen()) {
                    entity.getSession().close(reason);
                }
                connections.remove(entity);
            } catch (IOException e) {
                log.error("用户下线异常：", e);
            }
        }
        log.info("当前人数:" + connections.size());

    }

    /**
     * 获取WebSocketEntity
     */
    private static WebSocketEntity getUserEntity(String userId, String sendType) {
        WebSocketEntity entity = null;
        if (FlymeUtils.isEmpty(connections))
            return null;
        for (WebSocketEntity webSocketEntity : connections) {
            String uId = webSocketEntity.getUserId();
            String sType = webSocketEntity.getSendType();
            if (uId.equals(userId) && sType.equals(sendType)) {
                entity = webSocketEntity;
                break;
            }
        }
        return entity;
    }

    /**
     * 获取在线人数
     */
    private synchronized int getUserOnlineNum() {
        return connections.size();
    }

    @Override
    public void sendMessage(String userId, String sendType, EntityMap map) {
        pushMessage(map, userId, sendType);
    }

    @Override
    public void sendMessage(String userId, String sendType, Map map) {
        pushMessage(map, userId, sendType);
    }
}
